package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/19 9:26
 */
public class Flink08_Table_Window_Grouped_SQL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql("create table sensor(" +
                            "   id string, " +
                            "   ts bigint," +
                            "   vc int, " +
                            "   et as to_timestamp(from_unixtime(ts/1000)), " +  // 根据long型的数据计算出来一个时间戳类型, 用来计算水印
                            "   watermark for et AS et - interval '3' second " +  // 生成水印
                            ")with(" +
                            "   'connector' = 'filesystem'," +
                            "   'path' = 'input/sensor.txt', " +
                            "   'format' = 'csv'" +
                            ")");
    
        /*tEnv.sqlQuery("select " +
                          " id," +
                          " tumble_start(et, interval '3' second) w_start, " +
                          " tumble_end(et, interval '3' second) w_end, " +
                          " sum(vc) sum_vc " +
                          "from sensor " +
                          "group by id, tumble(et, interval '3' second)")
            .execute()
            .print();*/
    
       /* tEnv.sqlQuery("select " +
                          " id," +
                          " hop_start(et, interval '2' second, interval '5' second) w_start, " +
                          " hop_end(et, interval '2' second, interval '5' second) w_end, " +
                          " sum(vc) sum_vc " +
                          "from sensor " +
                          "group by id, hop(et, interval '2' second, interval '5' second)")// first interval是窗口的长度
            .execute()
            .print();*/
    
        tEnv.sqlQuery("select " +
                          " id," +
                          " session_start(et, interval '2' second) w_start, " +
                          " session_end(et, interval '2' second) w_end, " +
                          " sum(vc) sum_vc " +
                          "from sensor " +
                          "group by id, session(et, interval '2' second)")// first interval是窗口的长度
            .execute()
            .print();
       
    
    }
}
